package io.kiki.sba.registry.server.clientInterface.strategy.impl;


import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.metaserver.ProvideData;
import io.kiki.sba.registry.common.model.store.BaseInfo;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.common.model.store.Watcher;
import io.kiki.sba.registry.core.model.ReceivedConfigData;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.converter.ReceivedDataConverter;
import io.kiki.sba.registry.server.clientInterface.providedata.ConfigProvideDataWatcher;
import io.kiki.sba.registry.server.clientInterface.push.FirePushService;
import io.kiki.sba.registry.server.clientInterface.push.PushSwitchService;
import io.kiki.sba.registry.server.clientInterface.store.Watchers;
import io.kiki.sba.registry.server.clientInterface.strategy.RegistryStrategy;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;
import java.util.concurrent.TimeUnit;


public class DefaultRegistryStrategy implements RegistryStrategy {
    private static final Logger logger = LoggerFactory.getLogger(DefaultRegistryStrategy.class);
    protected final WatcherScanDog watcherScanDog = new WatcherScanDog();
    private final Thread thread = ConcurrentUtils.createDaemonThread("watcher-scan-dog", watcherScanDog);
    @Autowired
    protected FirePushService firePushService;
    @Autowired
    protected Config config;
    @Autowired
    protected PushSwitchService pushSwitchService;
    @Autowired
    protected ConfigProvideDataWatcher configProvideDataWatcher;
    @Autowired
    protected Watchers sessionWatchers;

    @Override
    public void start() {
        thread.start();
    }

    @Override
    public void afterPublisherRegister(Publisher publisher) {
    }

    @Override
    public void afterSubscriberRegister(Subscriber subscriber) {
        if (pushSwitchService.canIpPushLocal(subscriber.getClientUrl().getIp())) {
            firePushService.fireOnRegister(subscriber);
        }
    }

    private boolean checkWatcherVersion(Watcher watcher) {
        return watcher.getClientVersion() == BaseInfo.ClientVersion.StoreData;
    }

    @Override
    public void afterWatcherRegister(Watcher watcher) {
        if (!checkWatcherVersion(watcher)) {
            logger.warn("unsupported watch:{}, clientVersion={}", watcher.getDataInfoId(), watcher.getClientVersion());
            return;
        }
        if (config.isWatchConfigEnable()) {
            configProvideDataWatcher.watch(watcher.getDataInfoId());
        }
        if (!pushSwitchService.canIpPushLocal(watcher.getClientUrl().getIp())) {
            return;
        }
        ProvideData provideData = null;
        if (config.isWatchConfigEnable()) {
            // the provideData maybe exist
            provideData = configProvideDataWatcher.get(watcher.getDataInfoId());
        } else {
            // just push null content provideData when disable
            provideData = new ProvideData(null, watcher.getDataInfoId(), null);
        }
        if (provideData != null) {
            ReceivedConfigData data = ReceivedDataConverter.createReceivedConfigData(watcher, provideData);
            firePushService.fireOnWatcher(watcher, data);
        }
    }

    boolean processWatchWhenWatchConfigDisable(Watcher watcher) {
        if (watcher.hasPushed()) {
            return false;
        }
        ProvideData provideData = new ProvideData(null, watcher.getDataInfoId(), null);
        ReceivedConfigData receivedConfigData = ReceivedDataConverter.createReceivedConfigData(watcher, provideData);
        firePushService.fireOnWatcher(watcher, receivedConfigData);
        return true;
    }

    boolean processWatchWhenWatchConfigEnable(Watcher watcher) {
        final String dataInfoId = watcher.getDataInfoId();
        final ProvideData provideData = configProvideDataWatcher.get(dataInfoId);
        if (provideData == null) {
            logger.warn("[WatcherNoData]{}", dataInfoId);
            return false;
        }
        ReceivedConfigData receivedConfigData = null;
        if (!watcher.hasPushed()) {
            // first push
            receivedConfigData = ReceivedDataConverter.createReceivedConfigData(watcher, provideData);
        } else {
            Long v = provideData.getVersion();
            if (v != null && watcher.getPushedVersion() < v) {
                receivedConfigData = ReceivedDataConverter.createReceivedConfigData(watcher, provideData);
            }
        }
        if (receivedConfigData != null) {
            firePushService.fireOnWatcher(watcher, receivedConfigData);
            return true;
        }
        return false;
    }

    boolean processWatch(Watcher watcher, boolean watchEnable) {
        return watchEnable ? processWatchWhenWatchConfigEnable(watcher) : processWatchWhenWatchConfigDisable(watcher);
    }

    Tuple<Set<String>, List<Watcher>> filter() {
        List<Watcher> watcherList = new LinkedList<>(sessionWatchers.getDataList());
        if (CollectionUtils.isEmpty(watcherList)) {
            return null;
        }
        Set<String> dataInfoIds = new HashSet<>(1024);
        Iterator<Watcher> iterator = watcherList.iterator();
        while (iterator.hasNext()) {
            Watcher watcher = iterator.next();
            if (checkWatcherVersion(watcher)) {
                dataInfoIds.add(watcher.getDataInfoId());
            } else {
                iterator.remove();
            }
        }
        return Tuple.of(dataInfoIds, watcherList);
    }

    @Override
    public void afterPublisherUnRegister(Publisher publisher) {
    }

    @Override
    public void afterSubscriberUnRegister(Subscriber subscriber) {
    }

    @Override
    public void afterWatcherUnRegister(Watcher watcher) {
    }

    final class WatcherScanDog extends LoopExecuteTask {

        @Override
        public void _execute_() {
            Tuple<Set<String>, List<Watcher>> tuple = filter();
            if (tuple == null) {
                return;
            }
            final boolean watchConfigEnable = config.isWatchConfigEnable();
            if (watchConfigEnable) {
                configProvideDataWatcher.refreshWatch(tuple.o1);
            }

            int count = 0;
            for (Watcher watcher : tuple.o2) {
                if (processWatch(watcher, watchConfigEnable)) {
                    count++;
                }
            }
            logger.info("fire watchers:{}", count);
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(config.getScanWatcherIntervalMillis(), TimeUnit.MILLISECONDS);
        }
    }
}
